package cn.doitedu.rtdw.data_sync;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/24
 * @Desc: 学大数据，上多易教育
 *    业务库的用户注册信息表数据，同步到hbase
 **/
public class Job1_用户注册信息同步到hbase {
    public static void main(String[] args) {


        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 1. 将物理数据源（业务库的用户注册信息表binlog），映射成逻辑表
        tenv.executeSql(
                " create table ums_member_mysql(       "+
                        "   id    bigint,                       "+
                        " 	member_level_id bigint,             "+
                        " 	username string,                    "+
                        " 	phone string,                       "+
                        " 	status int,                         "+
                        " 	create_time timestamp(3),           "+
                        " 	gender int,                         "+
                        " 	birthday date,                      "+
                        " 	city string,                        "+
                        " 	job string,                         "+
                        " 	source_type int,                    "+
                        " 	modify_time timestamp(3),           "+
                        " 	primary key(id) not enforced        "+
                        " ) with (                              "+
                        "     'connector' = 'mysql-cdc',        "+
                        "     'hostname' = 'doitedu',           "+
                        "     'port' = '3306',                  "+
                        "     'username' = 'root',              "+
                        "     'password' = 'root',              "+
                        "     'database-name' = 'realtimedw',   "+
                        "     'table-name' = 'ums_member'       "+
                        " )                                     "
        );



        // 2.将目标物理数据源（hbase），映射成逻辑表
        tenv.executeSql(
                " create table dim_user_info_hbase(                                        "+
                        "    username string,                                                    "+
                        " 	 f1 row<id bigint,member_level_id bigint,phone string,status int,       "+
                        " 	       create_time timestamp(3), gender int,birthday date,city string, "+
                        " 		   job string,source_type int,modify_time timestamp(3)>            "+
                        " ) with (                                                                 "+
                        "  'connector' = 'hbase-2.2',                                              "+
                        "  'table-name' = 'dim_user_info',                                         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'                                     "+
                        " )                                                                        "
        );


        // 3.在逻辑表之间开发运算sql
        tenv.executeSql(
                "insert into dim_user_info_hbase                             "+
                        " select                                                      "+
                        "    username,                                                "+
                        "    row(id,member_level_id,phone,status,create_time,         "+
                        "    gender,birthday,city,job,source_type,modify_time ) as f1 "+
                        " from ums_member_mysql                                       "
        );

    }
}
